package pool

import (
	"github.com/618lf/swakx-go/pkg/future"
	"github.com/618lf/swakx-go/pkg/gofunc"
	"github.com/618lf/swakx-go/pkg/tools/panics"
	"math/rand"
	"runtime/debug"
	"sync"
	"sync/atomic"
)

/**
 *  一种可以控制并发数的并发模型
 */

type taskQueue struct {
	wp      *orderPool
	tasks   chan func()
	lock    sync.Mutex
	running bool
}

func (t *taskQueue) in() chan<- func() {
	return t.tasks
}

func (t *taskQueue) safeRun(f func()) {
	defer func() {
		if r := recover(); r != nil {
			if log != nil {
				log.Errorf("goroutine panic: %v\n%s", r, string(debug.Stack()))
			}
		}
	}()

	// execute task
	f()
}

func (t *taskQueue) popOne() func() {
	select {
	case <-t.wp.done:
		if length := len(t.tasks); length > 0 {
			log.Errorf("exit now while its task buffer length %d is greater than 0", length)
		}
		return nil

	case f, _ := <-t.tasks:
		return f
	default:
		return nil
	}
}

func (t *taskQueue) runAll() {
	defer func() {
		if r := recover(); r != nil {
			if log != nil {
				log.Errorf("goroutine panic: %v\n%s", r, string(debug.Stack()))
			}
		}
	}()

	for {
		t.lock.Lock()
		f := t.popOne()
		if f == nil {
			t.running = false
			t.lock.Unlock()
			return
		}
		t.running = true
		t.lock.Unlock()

		t.safeRun(f)
	}
}

func (t *taskQueue) run() error {
	pool := t.wp.workerPool
	t.lock.Lock()
	defer t.lock.Unlock()

	if !t.running {
		// 必须同步等待
		return pool.Go(func() {
			t.runAll()
		})
	}

	// 顺序的运行
	return nil
}

type OrderPool interface {
	WorkerPool
}

type orderPool struct {
	OrderPool
	idx          uint32
	workerPool   WorkerPool
	queues       int
	taskQueues   []taskQueue
	once         sync.Once
	done         chan struct{}
	panicHandler func(interface{})
}

func (wp *orderPool) selectQueue(t func()) *taskQueue {

	var (
		index uint32
		queue taskQueue
	)

	index = atomic.AddUint32(&wp.idx, 1) % uint32(wp.queues)
	queue = wp.taskQueues[index]

	// add task to queue
	select {
	case <-wp.done:
		return nil
	case queue.in() <- t:
		return &queue
	default:
	}

	// put the task to a random queue with a maximum of len(e.taskQueues)/2 attempts
	for i := 0; i < wp.queues/2; i++ {
		queue = wp.taskQueues[rand.Intn(wp.queues)]
		select {
		case queue.in() <- t:
			return &queue
		default:
			continue
		}
	}

	// wait for add success
	queue = wp.taskQueues[index]
	queue.in() <- t

	// add success
	return &queue
}

func (wp *orderPool) Go(f func()) error {

	// select and add task to queue
	queue := wp.selectQueue(f)
	if queue == nil {
		return Error_Executor_Closed
	}

	// run in queue
	return queue.run()
}

func (wp *orderPool) Execute(runnable gofunc.Runnable) error {
	return wp.Go(func() {
		defer panics.CatchPanicWithHandleError(func(err error) {
			if wp.panicHandler != nil {
				wp.panicHandler(err)
			}
		})
		runnable()
	})
}

func (wp *orderPool) Submit(runnable gofunc.Runnable) future.Future {
	f := future.NewFuture()
	err := wp.Go(func() {
		defer panics.CatchPanicWithHandleError(func(err error) {
			f.CompleteExceptionally(err)
			if wp.panicHandler != nil {
				wp.panicHandler(err)
			}
		})

		// run
		runnable()

		// complete future
		f.Complete(nil)
	})

	// add task error
	if err != nil {
		f.CompleteExceptionally(err)
	}

	// return future
	return f
}

func (wp *orderPool) Promise(supplier gofunc.Supplier) future.Promise {
	p := future.NewPromise(func(resolve future.Resolve, reject future.Reject) {
		err := wp.Go(func() {
			defer panics.CatchPanicWithHandleError(func(err error) {
				reject(err)
				if wp.panicHandler != nil {
					wp.panicHandler(err)
				}
			})
			res := supplier()
			resolve(res)
		})

		// add task error
		if err != nil {
			reject(err)
		}
	})

	return p
}

func NewOrderPool(parallel int, cap int, pool WorkerPool) *orderPool {

	if parallel <= 0 {
		parallel = 1
	}

	if cap <= 0 {
		cap = 1024
	}

	// init taskQueues
	var taskQueues []taskQueue
	taskQueues = make([]taskQueue, parallel)
	for i := 0; i < parallel; i++ {
		taskQueues[i] = taskQueue{
			tasks: make(chan func(), cap),
		}
	}

	// return order pool
	return &orderPool{
		queues:     parallel,
		workerPool: pool,
		taskQueues: taskQueues,
	}
}
